package com.learn.util;

import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class SetCheckPoint {

    public static void setCheckPoint(StreamExecutionEnvironment env , String ckpPath_GroupId_JobName) {

        System.setProperty("HADOOP_USER_NAME","hadoop");

        env.enableCheckpointing(1000 * 60 * 5);

        env.setStateBackend(new EmbeddedRocksDBStateBackend(true));

        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        //这个10.0.10.13:4007 是在hadoop那里的namenode地址
        env.getCheckpointConfig().setCheckpointStorage("hdfs://10.0.10.13:4007/user/flink/rocksdbcheckpoint/gtb-k2d/" + ckpPath_GroupId_JobName );

        //设置超时时间
        env.getCheckpointConfig().setCheckpointTimeout(1000 * 60 * 10);

        //设置ckp并发度
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

        //设置ckp的运行间隔，防止barrier一到就马上进行ck
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000);
        //程序重启后ckp清理ckp:
        // Delete externalized checkpoints on job cancellation 删除作业取消时的外部化检查点
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

    }

}
